package com.ruyuan.o2o.groupbuy.logcenter.controller;

import com.ruyuan.o2o.groupbuy.logcenter.es.*;
import com.ruyuan.o2o.groupbuy.logcenter.hbase.HBaseClient;
import lombok.extern.slf4j.Slf4j;
import org.apache.logging.log4j.util.Strings;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
import org.springframework.web.bind.annotation.*;

import javax.annotation.Resource;
import java.util.HashMap;
import java.util.Map;

/**
 * 日志中心
 * 消费kafka日志和处理前台查询请求
 *
 * @author ming qian
 */
@RestController
@Slf4j
@RequestMapping("/api")
public class LogController {

    /**
     * kafka组件
     */
    @Resource
    private KafkaListenerEndpointRegistry registry;

    /**
     * es配置
     */
    @Resource
    private ElasticsearchConfig esConfig;
    /**
     * es组件
     */
    @Resource
    private ElasticsearchClient esExecutors;

    /**
     * redis组件
     */
    @Resource
    private StringRedisTemplate stringRedisTemplate;

    /**
     * hbase组件
     */
    @Resource
    private HBaseClient hBaseClient;

    public static final String KAFKA_LISTENER_ID = "kafka-log-center";

    public static final String HBASE_TABLE_NAME = "LOG";

    public static final String INDEX_SUFFIX = "-index";

    public static final String LOG_REDIS_SERVER_LIST = "log-server";

    /**
     * 用于过滤日志，正常业务日志用“#”分割的消息长度
     */
    public static final Integer MESSAGE_LENGTH = 2;

    /**
     * 接入安装微服务
     * 建立ES日志索引、hbase日志表、启动kafka监听
     *
     * @param serverName 前台写入的单个服务名称
     */
    @PostMapping("/install")
    public Boolean install(String serverName) {
        try {
            //  创建索引，默认前台表单输入的名称就是索引名
            if (!esExecutors.createIndex(serverName.trim() + INDEX_SUFFIX)) {
                return false;
            }
            //  hbase建表，默认前台表单输入的名称就是表名
            if (!hBaseClient.createTable(HBASE_TABLE_NAME, new String[]{"logContent"})) {
                return false;
            }
            //  安装完成后启动消费kafka
            if (!registry.getListenerContainer(KAFKA_LISTENER_ID).isRunning()) {
                registry.getListenerContainer(KAFKA_LISTENER_ID).start();
            }

            //  redis保存接入日志中心的服务列表
            stringRedisTemplate.opsForHash().put(LOG_REDIS_SERVER_LIST, serverName, serverName);

            log.info("已启动服务【{}】的kafka监听程序", serverName);
            return true;
        } catch (Exception e) {
            throw new RuntimeException(e.getMessage());
        }
    }

    /**
     * 日志中心的核心逻辑：
     * 接收kafka topic消息 、 es保存除自定义日志外的全部数据、hbase保存完整message
     * ${customize.kafka.consumer.topics}：指定的接收topic
     *
     * @param message 主机名, IP, 微服务名称, 日期, 日志级别, 执行类(行号), 调用链路id, 业务id#自定义日志内容
     */
    @KafkaListener(id = KAFKA_LISTENER_ID, topics = "${customize.kafka.consumer.topics}")
    public void message(String message) {
        //  message先用","分割，得到各个字段
        String[] messageSplit = message.split(",".trim());

        //  用“,”分割得出日志字段
        String hostname = messageSplit[0].trim();
        String ip = messageSplit[1].trim();
        String serverName = messageSplit[2].trim();
        String date = messageSplit[3].trim();
        String level = messageSplit[4].trim();
        String className = messageSplit[5].trim();
        String invocationId = messageSplit[6].trim();

        //  最后一个元素是日志格式中的自定义日志内容
        String[] logSplit = messageSplit[messageSplit.length - 1].split("#");

        //  用“#”分割长度是2 （业务id#自定义日志内容）
        String businessId = "";
        if (logSplit.length == MESSAGE_LENGTH) {
            //  业务id
            businessId = logSplit[0].trim();
        }
        if (!stringRedisTemplate.opsForHash().hasKey(LOG_REDIS_SERVER_LIST, serverName.trim())) {
            log.error("当前服务：【{}】没有接入日志中心", serverName);
            return;
        }
        //  控制台打印看日志效果，生产上不打印此log，节省磁盘空间
        log.info("接收kafka完整消息：{}", message);

        try {
            //  插入elasticsearch
            Map<String, String> esMap = new HashMap<>(16);
            esMap.put("hostname", hostname);
            esMap.put("ip", ip);
            esMap.put("serverName", serverName);
            esMap.put("date", date);
            esMap.put("level", level);
            esMap.put("className", className);
            esMap.put("businessId", businessId);
            esMap.put("invocationId", invocationId);

            IndexRequest request = new IndexRequest();
            request.index(serverName + "-index").source(esMap).id();
            IndexResponse indexResponse = esConfig.esClient().index(request, RequestOptions.DEFAULT);
            String documentId = indexResponse.getId();

            //  插入hbase
            hBaseClient.insertHbase(HBASE_TABLE_NAME, message, documentId);

        } catch (
                Exception e) {
            e.printStackTrace();
            throw new RuntimeException("数据保存失败");
        }

    }

    /**
     * 查询日志信息列表
     *
     * @param page            页数    默认1
     * @param limit           条数    默认10
     * @param hostname        主机名
     * @param ip              ip地址
     * @param businessId      业务id
     * @param invocationIdStr 调用链路id
     * @return 查询结果
     */
    @PostMapping("/list")
    public SearchResultVo listPage(Integer page, Integer limit,
                                   String hostname, String ip, String businessId, String invocationIdStr) {
        try {
            SearchParameters searchParameters = new SearchParameters();
            BoolQueryBuilder boolQueryBuilder = new BoolQueryBuilder();

            //  参数校验
            if (page != null) {
                searchParameters.setFrom(page);
            }
            if (limit != null) {
                searchParameters.setSize(limit);
            }
            if (Strings.isNotBlank(hostname)) {
                boolQueryBuilder.must(QueryBuilders.termQuery(
                        "hostname", hostname.trim()));
            }
            if (Strings.isNotBlank(ip)) {
                boolQueryBuilder.must(QueryBuilders.termQuery(
                        "ip", ip.trim()));
            }
            if (Strings.isNotBlank(businessId)) {
                boolQueryBuilder.must(QueryBuilders.termQuery(
                        "businessId", businessId.trim()));
            }
            if (Strings.isNotBlank(invocationIdStr)) {
                boolQueryBuilder.must(QueryBuilders.termQuery(
                        "invocationId", invocationIdStr.trim()));
            }
            //  查询ES
            return esExecutors.search(searchParameters, boolQueryBuilder);

        } catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException("分页查询ES数据失败");
        }
    }

}

